package com.example.mq.mqserver;

import com.example.mq.common.MqException;
import com.example.mq.common.Router;
import com.example.mq.mqserver.core.*;
import com.example.mq.mqserver.datacenter.IDiskDataCenter;
import com.example.mq.mqserver.datacenter.IMemoryDataCenter;
import com.example.mq.mqserver.datacenter.Impl.DiskDataCenterImpl;
import com.example.mq.mqserver.datacenter.Impl.MemoryDataCenterImpl;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 用这个类表示一个虚拟主机，实现业务逻辑的整合
 */
@Slf4j
public class VirtualHost {
    //  为了防止不同主机的 交换机名或队列名重复的情况，统一在前面加上虚拟机名作为前缀
    private final String virtualHostName;
    private final IDiskDataCenter diskDataCenter = new DiskDataCenterImpl();
    private final IMemoryDataCenter memoryDataCenter = new MemoryDataCenterImpl();
    private final ConsumerManager consumerManager = new ConsumerManager(this);

    //  为了使锁的粒度更小，创建不同的锁对象
    private final Object exchangeLocker = new Object();
    private final Object queueLocker = new Object();

    public VirtualHost(String virtualHostName) {
        this.virtualHostName = virtualHostName;
        diskDataCenter.init();
        try {
            memoryDataCenter.recovery(diskDataCenter);
        } catch (IOException | ClassNotFoundException e) {
            log.warn("恢复数据失败! ");
            e.printStackTrace();
        }

    }

    public boolean exchangeDeclare(String exchangeName, ExchangeType exchangeType, boolean durable, boolean autoDelete, Map<String,Object> arguments) {
        //  重命名交换机
        exchangeName = virtualHostName + exchangeName;
        try {
            synchronized (exchangeLocker) {
                Exchange existsExchange = memoryDataCenter.getExchange(exchangeName);
                if(existsExchange != null) {
                    throw new MqException("交换机已存在! ");
                }

                Exchange exchange = new Exchange(exchangeName,exchangeType,durable,autoDelete);
                exchange.setArguments(arguments);
                if(durable) {
                    diskDataCenter.insertExchange(exchange);
                }
                memoryDataCenter.insertExchange(exchange);
                log.info("创建交换机成功. exchangeName=" + exchangeName);
                return true;
            }
        } catch (MqException e) {
            log.warn("创建交换机失败! ");
            return false;
        }

    }
    public boolean exchangeDelete(String exchangeName) {
        exchangeName = virtualHostName + exchangeName;
        try {
            synchronized (exchangeLocker) {
                Exchange existsExchange = memoryDataCenter.getExchange(exchangeName);
                if(existsExchange == null) {
                    throw new MqException("要删除的交换机不存在! ");
                }

                if(existsExchange.isDurable()) {
                    diskDataCenter.deleteExchange(existsExchange);
                }
                memoryDataCenter.deleteExchange(exchangeName);
                log.info("删除交换机成功. exchangeName=" + exchangeName);
                return true;
            }
        } catch (MqException e) {
            log.warn("删除交换机失败! ");
            e.printStackTrace();
            return false;
        }
    }

    public boolean queueDeclare(String queueName,boolean durable,boolean exclusive,boolean autoDelete,Map<String,Object> arguments) {
        queueName = virtualHostName + queueName;
        try {
            synchronized (queueLocker) {
                MSGQueue existsQueue = memoryDataCenter.getQueue(queueName);
                if(existsQueue != null) {
                    throw new MqException("队列已存在! ");
                }

                MSGQueue queue = new MSGQueue(queueName,durable,exclusive,autoDelete);
                queue.setArguments(arguments);
                if(durable) {
                    diskDataCenter.insertQueue(queue);
                }
                memoryDataCenter.insertQueue(queue);
                log.info("创建队列成功. queueName=" + queueName);
                return true;
            }
        } catch (MqException | IOException e) {
            log.warn("创建队列失败! ");
            e.printStackTrace();
            return false;
        }
    }
    public boolean queueDelete(String queueName) {
        queueName = virtualHostName + queueName;
        try {
            synchronized (queueLocker) {
                MSGQueue existsQueue = memoryDataCenter.getQueue(queueName);
                if(existsQueue == null) {
                    throw new MqException("要删除的队列不存在! ");
                }

                if(existsQueue.isDurable()) {
                    diskDataCenter.deleteQueue(existsQueue);
                }
                memoryDataCenter.deleteQueue(queueName);
                log.info("删除队列成功. queueName=" +queueName);
                return true;
            }
        } catch (Exception e) {
            log.warn("删除队列失败! ");
            e.printStackTrace();
            return false;
        }

    }

    public boolean queueBind(String exchangeName,String queueName,String bindingKey) {
        exchangeName = virtualHostName + exchangeName;
        queueName = virtualHostName + queueName;
        try {
            synchronized (exchangeLocker) {
                synchronized (queueLocker) {
                    Binding existsBinding = memoryDataCenter.getBinding(exchangeName, queueName);
                    if(existsBinding != null) {
                        throw new MqException("绑定已经存在! ");
                    }
                    Exchange existsExchange = memoryDataCenter.getExchange(exchangeName);
                    if(existsExchange == null) {
                        throw new MqException("要绑定的交换机不存在! ");
                    }
                    MSGQueue existsQueue = memoryDataCenter.getQueue(queueName);
                    if(existsQueue == null) {
                        throw new MqException("要绑定的队列不存在! ");
                    }
                    if(!Router.checkBindingKey(bindingKey)) {
                        throw new MqException("bindingKey 不合法! ");
                    }

                    Binding binding = new Binding(exchangeName,queueName,bindingKey);
                    if(existsExchange.isDurable() && existsQueue.isDurable()) {
                        diskDataCenter.insertBinding(binding);
                    }
                    memoryDataCenter.insertBinding(binding);
                    log.info("创建绑定成功. exchangeName=" + exchangeName + ",queueName=" + queueName);
                    return true;
                }
            }
        } catch (MqException e) {
            log.warn("创建绑定失败! ");
            e.printStackTrace();
            return false;
        }

    }
    public boolean queueUnbind(String exchangeName,String queueName) {
        exchangeName = virtualHostName + exchangeName;
        queueName = virtualHostName + queueName;
        try {
            synchronized (exchangeLocker) {
                synchronized (queueLocker) {
                    Binding existsBinding = memoryDataCenter.getBinding(exchangeName, queueName);
                    if(existsBinding == null) {
                        throw new MqException("要解除的绑定不存在! ");
                    }

                    //  直接删除硬盘上的数据，防止出现 交换机或队列没了，但绑定还在的情况
                    diskDataCenter.deleteBinding(existsBinding);
                    memoryDataCenter.deleteBinding(existsBinding);
                    log.info("解除绑定成功. exchangeName=" + exchangeName + ",queueName=" + queueName);
                    return true;
                }
            }
        } catch (MqException e) {
            log.warn("解除绑定失败! ");
            e.printStackTrace();
            return false;
        }

    }

    //  转发消息
    //  消息推送到对应交换机，再由交换机根据不同类型的转发规则，转发到对应的队列
    public boolean basicPublish(String exchangeName,BasicProperties basicProperties,String routingKey,byte[] body) {
        exchangeName = virtualHostName + exchangeName;
        try {
            //  防止转发消息时，交换机或者队列被删除，也要加个锁
            synchronized (exchangeLocker) {
                synchronized (queueLocker) {
                    Exchange exchange = memoryDataCenter.getExchange(exchangeName);
                    if(exchange == null) {
                        throw new MqException("转发的交换机不存在! ");
                    }
                    if(!Router.checkRoutingKey(routingKey)) {
                        throw new MqException("routingKey 不合法! ");
                    }

                    //  封装消息
                    Message message = Message.createMessageWithId(basicProperties,routingKey,body);
                    //  根据不同类型的交换机，实现不同的转发规则
                    if(exchange.getExchangeType() == ExchangeType.DIRECT) {
                        //  direct 交换机，routingKey = 队列名
                        String queueName = virtualHostName + routingKey;
                        MSGQueue queue = memoryDataCenter.getQueue(queueName);
                        if(queue == null) {
                            throw new MqException("转发的队列不存在! ");
                        }

                        sendMessage(queue,message);
                    } else {
                        //  fanout/topic 交换机都是转发至绑定的队列中，就一起处理了
                        //  fanout:转发到绑定的所有队列
                        //  topic:转发到匹配的队列
                        ConcurrentHashMap<String, Binding> bindingMap = memoryDataCenter.getBindings(exchangeName);
                        for(Map.Entry<String,Binding> entry : bindingMap.entrySet()) {
                            String queueName = entry.getKey();
                            Binding binding = entry.getValue();
                            MSGQueue queue = memoryDataCenter.getQueue(queueName);
                            if(queue == null) {
                                throw new MqException("转发的队列不存在! ");
                            }
                            if(!Router.route(exchange.getExchangeType(),binding.getBindingKey(),routingKey)) {
                                //  当前队列不匹配，继续遍历其他队列
                                continue;
                            }
                            //  要把消息克隆再发送，避免消息中心只有一份
                            Message cloneMessage = Message.createMessageWithId(basicProperties,routingKey,body);
                            sendMessage(queue,cloneMessage);
                        }
                    }

                    log.info("消息转发成功. messageId=" + message.getMessageId());
                    return true;
                }
            }
        } catch (Exception e) {
            log.warn("转发消息失败! ");
            e.printStackTrace();
            return false;
        }

    }
    private void sendMessage(MSGQueue queue, Message message) throws IOException, InterruptedException {
        if(message.getDeliverMode() == 0x2) {
            diskDataCenter.sendMessage(queue,message);
        }
        memoryDataCenter.offerMessage(queue.getName(),message);
        //  提醒消费者，可以进行消费
        consumerManager.notifyConsumer(queue.getName());
    }

    //  订阅队列
    //  具体就是 把消费者加入到队列的消费者集合中
    public boolean basicConsume(String consumerTag,String queueName,boolean autoAck,Consume consume) {
        queueName = virtualHostName + queueName;

        try {
            consumerManager.addConsumer(consumerTag,queueName,autoAck,consume);
            log.info("订阅队列成功. ");
            return true;

        } catch (MqException e) {
            log.warn("订阅队列失败! ");
            e.printStackTrace();
            return false;
        }

    }

    //  手动应答
    public boolean basicAck(String queueName,String messageId) {
        queueName = virtualHostName + queueName;
        try {
            MSGQueue queue = memoryDataCenter.getQueue(queueName);
            if(queue == null) {
                throw new MqException("队列不存在! ");
            }
            Message message = memoryDataCenter.getMessage(messageId);
            if(message == null) {
                throw new MqException("消息不存在! ");
            }

            if(message.getDeliverMode() == 0x2) {
                diskDataCenter.deleteMessage(queue,message);
            }
            memoryDataCenter.removeMessage(message.getMessageId());
            memoryDataCenter.removeMessageWaitAck(queue.getName(),message.getMessageId());

            log.info("手动应答成功. ");
            return true;
        } catch (IOException | MqException | ClassNotFoundException e) {
            log.warn("手动应答失败！ ");
            e.printStackTrace();
            return false;
        }

    }



    public String getVirtualHostName() {
        return virtualHostName;
    }
    public IDiskDataCenter getDiskDataCenter() {
        return diskDataCenter;
    }
    public IMemoryDataCenter getMemoryDataCenter() {
        return memoryDataCenter;
    }
    public ConsumerManager getConsumerManager() {
        return consumerManager;
    }
}
